package src

import (
	"context"
	"encoding/json"
	"log"
	"os"
	"strings"
	"sync"
	"time"

	"gitee.com/haifengat/goctp"
	ctp "gitee.com/haifengat/goctp/lnx"

	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"

	"github.com/go-redis/redis/v8"

	"gorm.io/driver/postgres"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
	"gorm.io/gorm/schema"
)

var (
	tradeFront, quoteFront, loginInfo, brokerID, investorID, password, appID, authCode string

	instLastMin         sync.Map // 合约:map[string]interface{},最后1分钟数据
	mapInstrumentStatus sync.Map // 合约交易状态

	rdb *redis.Client   // redis 连接
	ctx context.Context // redis 上下文

	actionDay     string   // 交易日起始交易日期
	actionDayNext string   // 交易日起始交易日期-下一日
	products      []string // 需要接收行情的品种（大写）

	t = ctp.NewTrade()
	q = ctp.NewQuote()

	GORMpg *gorm.DB
)

// NewRealMd realmd 实例
func init() {
	// 环境变量读取,赋值
	var tmp string
	if tmp = os.Getenv("tradeFront"); tmp == "" {
		logrus.Fatal("未配置环境变量：tradeFront")
	}
	tradeFront = tmp
	if tmp = os.Getenv("quoteFront"); tmp == "" {
		logrus.Fatal("未配置环境变量: quoteFront")
	}
	quoteFront = tmp
	if tmp = os.Getenv("loginInfo"); tmp == "" {
		logrus.Fatal("未配置环境变量: loginInfo")
	}
	loginInfo = tmp

	fs := strings.Split(loginInfo, "/")
	brokerID, investorID, password, appID, authCode = fs[0], fs[1], fs[2], fs[3], fs[4]
	if !strings.HasPrefix(tradeFront, "tcp://") {
		tradeFront = "tcp://" + tradeFront
	}
	if !strings.HasPrefix(quoteFront, "tcp://") {
		quoteFront = "tcp://" + quoteFront
	}

	var redisAddr = ""
	if tmp = os.Getenv("redisAddr"); tmp == "" {
		logrus.Error("未配置环境变量: redisAddr, 不能订阅实时行情.")
	} else {
		redisAddr = tmp

		logrus.Info("redis: ", redisAddr)
		rdb = redis.NewClient(&redis.Options{
			Addr:         redisAddr,
			Password:     "",  // no password set
			DB:           0,   // use default DB
			PoolSize:     100, // 连接池最大socket连接数，默认为4倍CPU数， 4 * runtime.NumCPU
			MinIdleConns: 10,  //在启动阶段创建指定数量的Idle连接，并长期维持idle状态的连接数不少于指定数量；
			//超时
			DialTimeout:  5 * time.Second, //连接建立超时时间，默认5秒。
			ReadTimeout:  3 * time.Second, //读超时，默认3秒， -1表示取消读超时
			WriteTimeout: 3 * time.Second, //写超时，默认等于读超时
			PoolTimeout:  3 * time.Second, //当所有连接都处在繁忙状态时，客户端等待可用连接的最大等待时长，默认为读超时+1秒
		})
		ctx = context.Background()
		pong, err := rdb.Ping(ctx).Result()
		if err != nil {
			logrus.Fatal(pong, err)
		}
	}

	pgMin := os.Getenv("pgMin")
	if pgMin == "" {
		logrus.Warn("未配置 pgMin, 收盘后将不入库！")
	} else {
		logrus.Info("postgres :", pgMin)
		newLogger := logger.New(
			log.New(os.Stdout, "\r\n", log.LstdFlags), // io writer
			logger.Config{
				SlowThreshold:             5000 * time.Millisecond, // 慢 SQL 阈值, 200ms
				LogLevel:                  logger.Warn,             // 日志级别
				IgnoreRecordNotFoundError: true,                    // 忽略ErrRecordNotFound（记录未找到）错误
				Colorful:                  true,                    // 彩色打印
			},
		)
		var schemaName = "futurn" // 模式
		var err error
		GORMpg, err = gorm.Open(postgres.Open(pgMin), &gorm.Config{
			SkipDefaultTransaction: true, // 跳过事务一致性检查,默认false,设为 true可提升性能,但不保证数据准确性
			NamingStrategy: schema.NamingStrategy{
				TablePrefix:   schemaName + ".t_", // 表名前置 User => t_users
				SingularTable: true,               // 是不表名复数, false:T_users  true:T_user
			},
			NowFunc:                                  func() time.Time { return time.Now() }, // 替换默认 CreateAt UpdateAt 的时间创建函数
			DisableForeignKeyConstraintWhenMigrating: true,                                   // 禁止创建物理外键, 推荐设为true, 因物理外键<非常>影响性能, 多用逻辑外键(代码中体现外键关系)
			Logger:                                   newLogger,
		})
		if err != nil {
			logrus.Fatal("pgMin 配置错误:", err)
		}
		err = GORMpg.Exec("CREATE SCHEMA IF NOT EXISTS " + schemaName + " AUTHORIZATION postgres;").Error
		if err != nil {
			logrus.Fatal(err)
		}
		GORMpg.AutoMigrate(&Bar{})
	}

	tmp = os.Getenv("products")
	if len(tmp) > 0 {
		products = strings.Split(tmp, ",")
		logrus.Info("products: ", tmp)
	}
}

func InsertPg() (err error) {
	// 取交易日
	tradingDay, err := rdb.HGet(ctx, "tradingday", "curday").Result()
	if err != nil {
		return errors.Wrap(err, "get curday")
	}
	// 删除当前交易日数据(实现重复入库)
	err = GORMpg.Where(&Bar{TradingDay: tradingDay}).Delete(&Bar{}).Error
	if err != nil {
		return errors.Wrap(err, "delete")
	}
	// 取所有 key
	var bars = make([]*Bar, 0)
	insts, err := rdb.Keys(ctx, "*").Result()
	if err != nil {
		return errors.Wrap(err, "get redis keys")
	}
	// 按合约key入库
	for _, inst := range insts {
		if inst == "tradingday" {
			continue
		}
		var mins = []string{}
		if mins, err = rdb.LRange(ctx, inst, 0, -1).Result(); err != nil {
			logrus.Error("取redis数据错误:", inst, err)
			continue
		}
		for _, bsMin := range mins {
			var bar Bar
			if err = json.Unmarshal([]byte(bsMin), &bar); err != nil {
				logrus.Error("解析bar错误:", bar, " ", err)
				continue
			}
			bars = append(bars, &bar)
			if len(bars) > 100000 { // 10W
				err = GORMpg.CreateInBatches(bars, 2000).Error
				if err != nil {
					return errors.Wrap(err, "create batch")
				}
				bars = make([]*Bar, 0)
			}
		}
	}
	err = GORMpg.CreateInBatches(bars, 2000).Error
	if err != nil {
		return errors.Wrap(err, "create batch")
	}
	return
}

func Run724() {
	startTrade()

	tick := time.NewTicker(1 * time.Minute)
	for range tick.C {
		hhmm := time.Now().Local().Format("15:04")
		if t.IsLogin { // 收盘后退出
			isContinue := false
			t.InstrumentStatuss.Range(func(key, value interface{}) bool {
				if value.(*goctp.InstrumentStatus).InstrumentStatus == goctp.InstrumentStatusContinous {
					isContinue = true
					return false
				}
				return true
			})
			if time.Now().Local().Minute()%15 == 0 { // 15 分钟显示一次
				logrus.Info(hhmm, " 交易中: ", isContinue)
			}
			if !isContinue {
				hour := time.Now().Local().Hour()
				if hour <= 3 || hour == 15 { // 夜盘结束&当日收盘
					logrus.Info("release at ", time.Now().Local())
					releaseTrade()
					releaseQuote()
				}
				if hour == 15 {
					logrus.Info("保存数据到 pg")
					if err := InsertPg(); err != nil {
						logrus.Error(err)
					}
				}
			}
		} else { // 定时连接
			if time.Now().Local().Minute()%15 == 0 { // 15 分钟显示一次
				logrus.Info(hhmm, " 接口关闭")
			}
			if hhmm == "08:40" || hhmm == "20:40" {
				logrus.Info("relogin at ", hhmm)
				startTrade()
			}
		}
	}
}
